package dyyx.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount2 {

	public static void main(String[] args) throws Exception {
		
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		 
        DataSet<String> text = env.fromElements(
            "hello java","hello tiger","hello tiger","cat cat cat","dog"
        );
        
        
 
        DataSet<Tuple2<String, Integer>> wordCounts = text
            .flatMap(new WordSplit())
            .groupBy(0)
            .sum(1);
 
        wordCounts.print();

	}
	
	private static class WordSplit implements FlatMapFunction<String, Tuple2<String, Integer>> {	 
	    @Override
	    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {    	
	        String[] words = value.toLowerCase().split(" "); 
	        for (String word : words) {
	        	word = word.trim();
	            if (word.length() > 0) {
	                out.collect(new Tuple2<String, Integer>(word, 1));
	            }   
	        }
	    }
	}

}
